import json

from kafka import KafkaConsumer

from common.api.rpc_connect_pool import init_connect_pool, delete_connect
from common.conf import KAFKA_BOOTSTRAP_SERVERS, SERVICE_LIST, SERVICE_TOPIC_REGISTER, \
    SERVICE_TOPIC_UNREGISTER
from common.conf import SERVICE_TIME_INTERVAL
from common.utils.etcd_util import get_etcd_client
from common.utils.session import redis_set
from common.utils.tasks import async_task


@async_task
def watch_service_changed(service_name, service_id):
    client = get_etcd_client()
    result = client.watch(key=service_id, timeout=SERVICE_TIME_INTERVAL + SERVICE_TIME_INTERVAL)

    if result.action == 'expire':
        try:
            # 获取service 列表
            services = json.loads(client.get(service_name).value)
        except BaseException:
            services = []
        if service_id in services:
            services.remove(service_id)
        client.set(key=service_name, value=json.dumps(services, ensure_ascii=False))


def register_service(service_id, service_name, host, port):
    client = get_etcd_client()
    print('register', service_id, service_name, host, port)

    try:
        # 获取service 列表
        services = json.loads(client.get(service_name).value)
    except BaseException as e:
        print(e)
        services = []

    if service_id not in services:
        services.append(service_id)

    info = json.dumps({
        'id': service_id,
        'name': service_name,
        'host': host,
        'port': port,
    })

    # 保存服务详情信息
    client.set(key=service_id, value=info, ttl=SERVICE_TIME_INTERVAL + 5)
    # 保存服务列表
    client.set(key=service_name, value=json.dumps(services))
    # 监听服务是否过期
    watch_service_changed(service_name=service_name, service_id=service_id)

    init_connect_pool(service_name=service_name, service_id=service_id, host=host, port=port)


def unregister_service(service_id, service_name):
    client = get_etcd_client()
    print('unregister', service_id, service_name, '------')

    try:
        # 获取service 列表
        services = json.loads(client.get(service_name).value)
    except BaseException as e:
        print(e)
        services = []
    if service_id in services:
        services.remove(service_id)

    try:
        client.delete(key=service_id)
    except BaseException:
        pass

    # 保存服务列表
    client.set(key=service_name, value=json.dumps(services))

    # rpc 连接池 删除连接记录
    delete_connect(service_name, service_id)


@async_task
def __listen_developer_log_status():
    client = get_etcd_client()
    k = 'developer_log_status'
    try:
        redis_set(key=k, value=client.get(k).value)
    except BaseException:
        pass
    while True:
        result = client.watch(k, timeout=6000)
        if result.action == 'set':
            redis_set(key=k, value=result.value)


@async_task
def __start_kafka_to_listen_services():
    service_consumer = KafkaConsumer(SERVICE_TOPIC_REGISTER,
                                     SERVICE_TOPIC_UNREGISTER,
                                     bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS)
    while True:
        for msg in service_consumer:
            content = eval(msg.value)
            if msg.topic == SERVICE_TOPIC_REGISTER:
                register_service(service_id=content['service_id'], service_name=content['service_name'],
                                 host=content['host'], port=content['port'])
            elif msg.topic == SERVICE_TOPIC_UNREGISTER:
                unregister_service(service_id=content['service_id'], service_name=content['service_name'])


def start_listen_service():
    __listen_developer_log_status()
    __start_kafka_to_listen_services()
